use std::{
    collections::HashMap,
    net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4},
    time::Duration,
};

use crate::shared::{AbortOnDropHandle, Packet, PacketType, MAX_UDP_BUFFER_SIZE};
use anyhow;
use jarust::jaconfig::JaConfig;
use jarust::japrotocol::EstablishmentProtocol;
use jarust::japrotocol::RTP;
use jarust_plugins::audio_bridge::messages::AudioBridgeJoinOptions;
use jarust_plugins::audio_bridge::AudioBridge;
use jarust_plugins::audio_bridge::{
    events::AudioBridgePluginEvent, handles::AudioBridgeHandle, messages::AudioBridgeCreateOptions,
};
use tokio::{
    net::{unix::pipe::Receiver, UdpSocket},
    sync::mpsc,
};

pub struct ConfBridge {
    cid: String,
    crypto_key: String,
    sender: mpsc::Sender<CtrlMsg>,
    handle: AbortOnDropHandle<()>,
}
pub struct ConfBridgeChannel {
    sender: mpsc::UnboundedSender<Packet>,
}

enum CtrlMsg {
    AddChannel {
        pid: String,
        local_payload_type: u8,
        local_ssrc: u32,
        remote_payload_type: u8,
        remote_ssrc: Option<u32>,

        pkt_receiver: mpsc::UnboundedReceiver<Packet>,
        egress_pkt_sender: mpsc::Sender<TransportPacket>,
    },
    RemoveChannel {
        pid: String,
    },
}

struct ConfBridgeChannelContext {
    pid: String,
    sender: mpsc::UnboundedSender<ChannelCtrlMsg>,
    local_ssrc: u32,
    audio_level: i8, // last Client-to-Mixer Audio Level Indication
    handle: AbortOnDropHandle<()>,
}

enum ChannelCtrlMsg {
    Encode,
}

pub struct TransportPacket {
    pub typ: PacketType,
    pub len: usize,
    pub buf: [u8; MAX_UDP_BUFFER_SIZE],
}

impl ConfBridge {
    pub fn new(local_ip: &str, server_ip: &str, cid: &str, crypto_key: String) -> Self {
        let cid = cid.to_string();
        let (sender, receiver) = mpsc::channel(32);
        let handle = tokio::spawn(conf_bridge_loop(
            local_ip.to_string(),
            server_ip.to_string(),
            cid.clone(),
            crypto_key.clone(),
            receiver,
        ))
        .into();
        Self {
            cid,
            crypto_key,
            sender,
            handle,
        }
    }
    pub async fn create_channel(
        &self,
        pid: &str,
        egress_packet_sender: mpsc::Sender<TransportPacket>,
        local_payload_type: u8,
        local_ssrc: u32,
        remote_payload_type: u8,
        remote_ssrc: Option<u32>,
    ) -> Result<ConfBridgeChannel, i32> {
        let pid = pid.to_string();
        let (pkt_sender, pkt_receiver) = mpsc::unbounded_channel();
        let _ = self
            .sender
            .send(CtrlMsg::AddChannel {
                pid,
                local_payload_type,
                local_ssrc,
                remote_payload_type,
                remote_ssrc,
                pkt_receiver,
                egress_pkt_sender: egress_packet_sender,
            })
            .await
            .map_err(|_| -1)?;
        Ok(ConfBridgeChannel { sender: pkt_sender })
    }
    pub async fn remove_channel(&self, pid: &str) -> Result<(), i32> {
        self.sender
            .send(CtrlMsg::RemoveChannel {
                pid: pid.to_string(),
            })
            .await
            .map_err(|_| -1)
    }
}

impl ConfBridgeChannel {
    pub fn received_rtp(&self, packet: &[u8]) {
        self.send_packet(PacketType::Rtp, packet)
    }
    pub fn received_rtcp(&self, packet: &[u8]) {
        self.send_packet(PacketType::Rtcp, packet)
    }
    fn send_packet(&self, typ: PacketType, packet: &[u8]) {
        let remote_addr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0));
        let len = std::cmp::min(packet.len(), MAX_UDP_BUFFER_SIZE);
        let mut pkt = Packet {
            remote_addr,
            typ,
            len,
            buf: [0u8; MAX_UDP_BUFFER_SIZE],
        };
        pkt.buf[..len].copy_from_slice(&packet[..len]);
        let _ret = self.sender.send(pkt);
        // log::info!("send_packet len={}, ret={}", len, _ret.is_ok());
    }
}

async fn conf_bridge_loop(
    local_ip: String,
    server_ip: String,
    cid: String,
    crypto_key: String,
    mut receiver: mpsc::Receiver<CtrlMsg>,
) {
    // pid -> channel context
    let mut channel_map: HashMap<String, ConfBridgeChannelContext> = HashMap::new();

    let res = create_room(&server_ip, cid.parse().unwrap()).await;
    if let Err(err) = res {
        log::error!("create_room {} failed, error: {}", cid, err);
        // return;
    }

    loop {
        tokio::select! {
            Some(msg) = receiver.recv() => match msg {
                CtrlMsg::AddChannel { pid, local_payload_type, local_ssrc, remote_payload_type, remote_ssrc, egress_pkt_sender, pkt_receiver } => {
                    let (chan_sender, chan_receiver) = mpsc::unbounded_channel();
                    let handle = tokio::spawn(conf_bridge_channel_loop(
                        local_ip.clone(),
                        server_ip.clone(),
                        cid.clone(),
                        pid.clone(),
                        local_payload_type,
                        local_ssrc,
                        remote_payload_type,
                        remote_ssrc,
                        egress_pkt_sender,
                        pkt_receiver,
                        chan_receiver)).into();
                    channel_map.insert(pid.clone(), ConfBridgeChannelContext {
                        pid,
                        sender: chan_sender,
                        local_ssrc,
                        audio_level: -120,
                        handle
                    });
                },
                CtrlMsg::RemoveChannel { pid } => {
                    channel_map.remove(&pid);
                },
            },
            else => break,
        }
    }
    log::info!("conf_bridge_loop exit");
}

async fn conf_bridge_channel_loop(
    local_ip: String,
    server_ip: String,
    cid: String,
    pid: String,
    local_payload_type: u8,
    local_ssrc: u32,
    remote_payload_type: u8,
    remote_ssrc: Option<u32>,
    egress_pkt_sender: mpsc::Sender<TransportPacket>, // receive from network,
    mut pkt_receiver: mpsc::UnboundedReceiver<Packet>, // receiver from user, send out
    mut ctrl_receiver: mpsc::UnboundedReceiver<ChannelCtrlMsg>,
) {
    // log::info!("conf_bridge_channel_loop enter");
    let sock = UdpSocket::bind("0.0.0.0:0").await.unwrap();
    let local_addr = sock.local_addr().unwrap();

    let res = join_room(
        &server_ip,
        cid.parse().unwrap(),
        pid.parse().unwrap(),
        &local_ip,
        local_addr.port(),
    )
    .await;
    if let Err(err) = res {
        log::error!("create_room {} failed, error: {}", cid, err);
        return;
    }
    let (_handle, _event_receiver, remote_rtp_addr) = res.unwrap();
    let remote_rtp_addr = format!("{}:{}", remote_rtp_addr.ip, remote_rtp_addr.port);
    let mut buf = [0u8; 2000];
    sock.connect(&remote_rtp_addr).await.unwrap();

    loop {
        tokio::select! {
            Some(msg) = ctrl_receiver.recv() => match msg {
                ChannelCtrlMsg::Encode => {}
            },
            Some(pkt) = pkt_receiver.recv() => {
                let _ = sock.send(&pkt.buf[..pkt.len]).await;
            },
            _res = sock.recv(&mut buf) => {
                // log::info!("recv {:?}", res);
            },
            else => break,
        }
    }
    log::info!("conf_bridge_channel_loop exit");
}

async fn create_room(
    server_ip: &str,
    room: u64,
) -> anyhow::Result<(AudioBridgeHandle, mpsc::Receiver<AudioBridgePluginEvent>)> {
    let timeout = std::time::Duration::from_secs(10);
    // tracing::info!("Start polling events...");
    let server_url = format!("ws://{}:8188/ws", server_ip);
    let config = JaConfig::builder().url(&server_url).build();
    let mut connection = jarust::connect(config, jarust::jaconfig::TransportType::Ws).await?;
    let session = connection.create(10).await?;
    let (handle, event_receiver) = session.attach_audio_bridge().await?;

    let options = AudioBridgeCreateOptions {
        room: Some(room),
        allow_rtp_participants: Some(true),
        ..Default::default()
    };
    let _result = handle.create_room_with_config(options, timeout).await?;
    Ok((handle, event_receiver))
}

async fn join_room(
    server_ip: &str,
    room: u64,
    participant_id: u64,
    local_ip: &str,
    local_port: u16,
) -> anyhow::Result<(
    AudioBridgeHandle,
    mpsc::Receiver<AudioBridgePluginEvent>,
    RTP,
)> {
    let timeout = std::time::Duration::from_secs(10);
    // tracing::info!("Start polling events...");
    let server_url = format!("ws://{}:8188/ws", server_ip);
    let config = JaConfig::builder().url(&server_url).build();
    let mut connection = jarust::connect(config, jarust::jaconfig::TransportType::Ws).await?;
    let session = connection.create(10).await?;
    let (handle, mut event_receiver) = session.attach_audio_bridge().await?;

    let options = AudioBridgeJoinOptions {
        id: participant_id.into(),
        ..Default::default()
    };
    let protocol = EstablishmentProtocol::RTP(RTP {
        ip: local_ip.to_string(),
        port: local_port,
        payload_type: 96.into(),
        audiolevel_ext: None,
        fec: None,
    });
    let _result = handle
        .join_room(room, options, protocol.into(), timeout)
        .await
        .map_err(|e| {
            tracing::error!("join_room error: {}", e);
            e
        })?;

    if let Some(AudioBridgePluginEvent::JoinRoom {
        establishment_protocol: Some(EstablishmentProtocol::RTP(rtp)),
        ..
    }) = event_receiver.recv().await
    {
        return Ok((handle, event_receiver, rtp));
    } else {
        return Err(anyhow::anyhow!("join room {} failed.", room));
    }
}
